package com.zlm.realtime.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * Author: Harbour
 * Date: 2021-07-06 20:19
 * Desc: kafka工具类
 */
public class MyKafkaUtils {

    private final static String kafkaServer = "bootstrap.servers=hadoop201:9092,hadoop202:9092,hadoop203:9092";

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) {

        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);

        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), prop);
    }

}
